package com.junbao.message.infrastructure.queue.bootstrap;

import java.util.Calendar;
import java.util.Iterator;
import java.util.TimerTask;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.junbao.message.infrastructure.queue.AbstractTask;
import com.junbao.message.infrastructure.queue.Slot;
import com.junbao.message.infrastructure.queue.WheelQueue;

/**
 * @ClassName QueueScanTimer
 * @Description 类似钟表的秒针，队列是表盘，这里有个类似秒针的循环器，
 *              每秒循环一次；就类似秒针再走。
 * @date 2017-03-21
 * @version 1.0.0
 *
 */
public class QueueScanTimer extends TimerTask {
	private static final Logger LOGGER = LoggerFactory.getLogger(QueueScanTimer.class);
	
	/**环形队列*/
	private WheelQueue queue;

	private static ThreadFactory slotThreadFactory = new MyDefaultThreadFactory("slotThreadGroup");

	private static ThreadFactory taskThreadFactory = new MyDefaultThreadFactory("taskThreadGroup");
	
	/**处理每个槽位的线程，循环到这个槽位，立即丢到一个线程去处理，然后继续循环队列。*/
	private ThreadPoolExecutor slotPool = new ThreadPoolExecutor(10, 10,
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<Runnable>(), slotThreadFactory);
	
	/**处理每一个槽位中task集合的线程， 集合中的每个任务一个线程*/
	private ThreadPoolExecutor taskPool = new ThreadPoolExecutor(8, 40,
            0L, TimeUnit.MILLISECONDS,
            new ArrayBlockingQueue<Runnable>(1024), taskThreadFactory);
		
	public QueueScanTimer(WheelQueue queue) {
		this.queue = queue;
	}

	@Override
	public void run() {
		try {
			if (queue == null) {
				return;
			}
			Calendar calendar = Calendar.getInstance();
			//LOGGER.info("----------------------");
			/**当前时间秒数*/
			int currentSecond = calendar.get(Calendar.MINUTE) * 60 + calendar.get(Calendar.SECOND);
			//获取当前秒之前所有的solt节点
			Slot[] slots = queue.peek(currentSecond);
			LOGGER.debug("current solt:" + currentSecond);
			
			slotPool.execute(new SlotTask(currentSecond,slots));
		} catch (Exception e) {
			//这里一个槽位的屏蔽异常，继续执行。
			LOGGER.error(e.getMessage(), e);
		}
	}

	/**
	 * 槽位任务
	 * @author hongjian.liu
	 *
	 */
    final class SlotTask implements Runnable {
    	final ConcurrentLinkedQueue<AbstractTask> totalTasks;
    	final int currentSecond;
    	final Slot[] slots;
    	
		public SlotTask(int currentSecond,Slot[] slots) {
			this.currentSecond = currentSecond;
			this.slots = slots;
			totalTasks = new ConcurrentLinkedQueue<AbstractTask>();
			for (Slot slot : slots) {
				ConcurrentLinkedQueue<AbstractTask> tasks = slot.getTasks();
				totalTasks.addAll(tasks);
			}
		}

		public void run() {
			if (totalTasks == null) {
				return;
			}
			String taskId;
			Iterator<AbstractTask> it = totalTasks.iterator();
            while (it.hasNext()) {
            	AbstractTask task = it.next();
            	if (LOGGER.isDebugEnabled()) {
            		LOGGER.debug("running_current_solt:currentSecond={}, task={}, taskQueueSize={}",new Object[]{ currentSecond, task.toString(), totalTasks.size()});
            	}
                taskId = task.getId();
                /**
                 * 判断任务是否到达可执行时间
                 */
                if (task.taskAlreadyExecute()) {
                	 taskPool.execute(task);
                	 it.remove();
                	 queue.getTaskSlotMapping().remove(taskId);
				}else {
					continue;
				}
            }
		}
	}

   
    static class MyDefaultThreadFactory implements ThreadFactory {
        private final ThreadGroup group;
        private final AtomicInteger threadNumber = new AtomicInteger(1);
        private final String namePrefix;

        MyDefaultThreadFactory(String groupName) {
            SecurityManager s = System.getSecurityManager();
            group = (s != null) ? s.getThreadGroup() :Thread.currentThread().getThreadGroup();
            namePrefix = groupName + "-pool-" + "-thread-";
        }
        
        public Thread newThread(Runnable runnable) {
            Thread t = new Thread(group, runnable, namePrefix + threadNumber.getAndIncrement(), 0);
            t.setDaemon(true);
            if (t.getPriority() != Thread.NORM_PRIORITY) {
                t.setPriority(Thread.NORM_PRIORITY);
            }
            return t;
        }
    }

}
